// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package base

import (
	"context"
	"sync"
	"sync/atomic"
	"time"
)

const (
	DefaultBucketBatch       = 100
	DefaultBucketConcurrency = 3
	DefaultBucketFrequency   = 350
	DefaultBucketSize        = 3000
)

// Bucket
// is a component that stores data in memory for asynchronous processing.
type Bucket[T any] struct {
	ctx     context.Context
	cancel  context.CancelFunc
	handler func(list []T)
	list    []T
	mu      *sync.RWMutex

	name string

	concurrency, processing int32
	frequency               int
	size, batch             int
}

// NewBucket
// creates a new bucket instance for memory data storage.
//
//	return NewBucket("span", handler)
//	return NewBucket("logger", handler)
func NewBucket[T any](name string, handler func([]T)) *Bucket[T] {
	return (&Bucket[T]{
		handler: handler,
		list:    make([]T, 0),
		mu:      new(sync.RWMutex),
		name:    name,
	}).init()
}

// Add
// adds a list items into memory.
func (o *Bucket[T]) Add(items ...T) {
	if func(list []T) int {
		o.mu.Lock()
		defer o.mu.Unlock()
		o.list = append(o.list, list...)
		return len(o.list)
	}(items) >= o.size {
		o.pop()
	}
}

// Name
// returns the bucket name.
func (o *Bucket[T]) Name() string {
	return o.name
}

// SetBatch
// sets the batch size for each pop.
func (o *Bucket[T]) SetBatch(batch int) {
	if batch > 0 {
		o.batch = batch
	}
}

// SetConcurrency
// sets the concurrency for log or span reporter.
func (o *Bucket[T]) SetConcurrency(concurrency int32) {
	if concurrency > 0 {
		o.concurrency = concurrency
	}
}

// SetFrequency
// sets the frequency for log or span reporter.
func (o *Bucket[T]) SetFrequency(frequency int) {
	if frequency > 0 {
		o.frequency = frequency
	}
}

// SetSize
// sets the maximum count messages for memory storage.
func (o *Bucket[T]) SetSize(size int) {
	if size > 0 {
		o.size = size
	}
}

// Start
// starts the bucket handler in a goroutine.
func (o *Bucket[T]) Start() {
	o.ctx, o.cancel = context.WithCancel(context.Background())

	go func() {
		// Block
		// and pop data to report log or span when the bucket service
		// is exited.
		defer func() {
			for i := 0; i < int(o.concurrency); i++ {
				go o.pop()
			}
		}()

		// Create
		// ticker to pop data to report log or span.
		t := time.NewTicker(time.Duration(o.frequency) * time.Millisecond)

		// Block
		// and receive signal.
		for {
			select {
			case <-t.C:
				// Call
				// pop for a while with frequency setting.
				go o.pop()

			case <-o.ctx.Done():
				// Stop
				// by context cancelled.
				return
			}
		}
	}()
}

// Stop
// stops the bucket handler.
func (o *Bucket[T]) Stop() {
	if o.ctx != nil && o.ctx.Err() == nil {
		o.cancel()
	}
}

// Wait
// until memory data is empty and processes are done.
func (o *Bucket[T]) Wait() {
	// Cancel
	// context if context is not cancelled.
	if o.ctx != nil && o.ctx.Err() == nil {
		o.cancel()
	}

	// Return
	// if memory data is empty and running processes are done.
	if atomic.LoadInt32(&o.processing) == 0 && func() bool {
		o.mu.RLock()
		defer o.mu.RUnlock()
		return len(o.list) == 0
	}() {
		return
	}

	// Wait
	// 10ms and check again.
	time.Sleep(time.Millisecond * 10)
	o.Wait()
}

// +------------------------------------------------------------------------------+
// |                            Access methods                                    |
// +------------------------------------------------------------------------------+

func (o *Bucket[T]) init() *Bucket[T] {
	o.batch = DefaultBucketBatch
	o.concurrency = DefaultBucketConcurrency
	o.frequency = DefaultBucketFrequency
	o.processing = 0
	o.size = DefaultBucketSize
	return o
}

func (o *Bucket[T]) pop() {
	var (
		items []T
		total int
	)

	// Flow-Control.
	if concurrency := atomic.AddInt32(&o.processing, 1); concurrency > o.concurrency {
		atomic.AddInt32(&o.processing, -1)
		return
	}

	// Pop data from memory.
	o.mu.Lock()
	if total = len(o.list); total <= o.batch {
		// Bucket is empty.
		if total == 0 {
			o.mu.Unlock()
			atomic.AddInt32(&o.processing, -1)
			return
		}

		// Pop all data.
		items = o.list[:]
		o.list = make([]T, 0)
		o.mu.Unlock()
	} else {
		// Pop batch data from head.
		items = o.list[0:o.batch]
		o.list = o.list[o.batch:]
		o.mu.Unlock()
	}

	// Call reporter to deal popped data.
	o.handler(items)
	atomic.AddInt32(&o.processing, -1)

	// Continue
	// to do until done.
	o.pop()
}
